package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Demo10_LookupJoin {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 用户行为日志记录表，kafka连接器
        tenv.executeSql(
                "create table user_events_kafka(\n" +
                        "     uid int,\n" +
                        "     event_id string,\n" +
                        "     properties map<string,string>,\n" +
                        "     action_time bigint, \n" +
                        "     pt as proctime() \n" +
                        ") with (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'tpc-a',\n" +
                        "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                        "  'properties.group.id' = 'doit47-g2',\n" +
                        "  'scan.startup.mode' = 'latest-offset',\n" +
                        "  'value.format' = 'json',\n" +
                        "  'value.fields-include' = 'EXCEPT_KEY'\n" +
                        ")");


        // 映射mysql的用户信息表，作为lookup join的查找表用
        tenv.executeSql(
                "CREATE TABLE stu_info_mysql      (            \n" +
                "  id int,                                     \n" +
                "  name string,                                \n" +
                "  gender string,                               \n" +
                "  age  int,                                    \n" +
                "  score  float,                                \n" +
                "  primary key(id) not enforced              \n" +
                ") WITH (                                     \n" +
                "  'connector' = 'jdbc',                     \n" +
                "  'url' = 'jdbc:mysql://doitedu:3306/doit47', \n" +
                "  'table-name' = 'stu_info',                 \n" +
                "  'username' = 'root',                      \n" +
                "  'password' = 'root',                      \n" +
                "  'lookup.cache' = 'NONE',                      \n" +
                "  'lookup.partial-cache.expire-after-access' = '1 h'  \n" +
                ")");


        tenv.executeSql("SELECT\n" +
                "    e.*,\n" +
                "    u.*\n" +
                "from user_events_kafka e \n" +
                "join stu_info_mysql for system_time as of e.pt as u \n" +
                "on e.uid = u.id \n").print();

    }
}
